dimos.protocol.pubsub.impl.test_zenohpubsub.TestZenohPubSubBase::test_untyped_topic_with_dotted_segment_round_trips

Stack Traces | 0.505s run time

request = <SubRequest 'monitor_threads' for <Function test_untyped_topic_with_dotted_segment_round_trips>>

    @pytest.fixture(autouse=True)
    def monitor_threads(request):
        # Capture threads before test runs
        test_name = request.node.nodeid
        with _seen_threads_lock:
            _before_test_threads[test_name] = {
                t.ident for t in threading.enumerate() if t.ident is not None
            }
    
        yield
    
        with _seen_threads_lock:
            before = _before_test_threads.get(test_name, set())
            current = {t.ident for t in threading.enumerate() if t.ident is not None}
    
            # New threads are ones that exist now but didn't exist before this test
            new_thread_ids = current - before
    
            if not new_thread_ids:
                return
    
            # Get the actual thread objects for new threads
            new_threads = [
                t for t in threading.enumerate() if t.ident in new_thread_ids and t.name != "MainThread"
            ]
    
            # Filter out expected persistent threads that are shared globally
            # These threads are intentionally left running and cleaned up on process exit
            expected_persistent_thread_prefixes = [
                "Dask-Offload",
                # HuggingFace safetensors conversion thread - no user cleanup API
                # https://github..../transformers/issues/29513
                "Thread-auto_conversion",
            ]
            new_threads = [
                t
                for t in new_threads
                if not any(t.name.startswith(prefix) for prefix in expected_persistent_thread_prefixes)
            ]
    
            # Filter out threads we've already seen (from previous tests)
            truly_new = [t for t in new_threads if t.ident not in _seen_threads]
    
            # Mark all new threads as seen
            for t in new_threads:
                if t.ident is not None:
                    _seen_threads.add(t.ident)
    
            if not truly_new:
                return
    
            thread_names = [t.name for t in truly_new]
    
>           pytest.fail(
                f"Non-closed threads created during this test. Thread names: {thread_names}. "
                "Please look at the first test that fails and fix that."
            )
E           Failed: Non-closed threads created during this test. Thread names: ['Thread-2660 (pyo3-closure)']. Please look at the first test that fails and fix that.

before     = {140175489361600, 140185641771136}
current    = {140175489361600, 140176008935104, 140185641771136}
expected_persistent_thread_prefixes = ['Dask-Offload', 'Thread-auto_conversion']
new_thread_ids = {140176008935104}
new_threads = [<Thread(Thread-2660 (pyo3-closure), stopped 140176008935104)>]
request    = <SubRequest 'monitor_threads' for <Function test_untyped_topic_with_dotted_segment_round_trips>>
t          = <Thread(Thread-2660 (pyo3-closure), stopped 140176008935104)>
test_name  = '.../pubsub/impl/test_zenohpubsub.py::TestZenohPubSubBase::test_untyped_topic_with_dotted_segment_round_trips'
thread_names = ['Thread-2660 (pyo3-closure)']
truly_new  = [<Thread(Thread-2660 (pyo3-closure), stopped 140176008935104)>]

dimos/conftest.py:269: Failed